/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package maxent.mr.io;

import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Iterator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.util.ReflectionUtils;
import stanford.tagger.ReadDataTagged;

/**
 *
 * @author hadoop
 */
public class ReadDataTaggedUtil {

    HashMap<Integer, Integer> rdts = new HashMap<Integer,Integer>();
    ReadDataTagged rdt;

    public ReadDataTaggedUtil(ReadDataTagged rdt){
        this.rdt = rdt;
        init();
    }

    private void init() {

        for (int i=0; i<rdt.getSize();i++){
            rdts.put(i, 0);
        }
    }

    public void write(String uri){

        Configuration conf = new Configuration();
        FileSystem fs;
        Path path = new Path(uri);
        IntWritable key = new IntWritable();
        IntWritable value = new IntWritable();
        SequenceFile.Writer writer = null;

        try{

            fs = FileSystem.get(URI.create(uri), conf);
            writer = SequenceFile.createWriter(fs, conf, path, key.getClass(), value.getClass());

            Iterator<Integer> indexIt = rdts.keySet().iterator();
            while(indexIt.hasNext()){
                Integer index = indexIt.next();
                key.set(index);
                value.set(rdts.get(index));
                //System.out.printf("[%s]\tkey:%s\tvalue:%s\n",writer.getLength(),key, value);
                writer.append(key, value);

            }

        }catch(IOException ex){
            ex.printStackTrace();
        }
        finally{
            IOUtils.closeStream(writer);
        }

    }

    public static HashMap<Integer, Integer> read(String uri){

        HashMap<Integer, Integer> rdts = new HashMap<Integer, Integer>();

        Configuration conf = new Configuration();

        SequenceFile.Reader reader = null;

        try{
            FileSystem fs = FileSystem.get(conf);
            FileStatus[] inputFiles = fs.listStatus(new Path(uri));


            for (int i = 0; i < inputFiles.length; i++) {
                // don't process the log files
                if (!inputFiles[i].getPath().getName().equalsIgnoreCase("_logs")) {
                    reader = new SequenceFile.Reader(fs, inputFiles[i].getPath(), conf);

                    IntWritable key = (IntWritable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
                    IntWritable value = (IntWritable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
                    long position = reader.getPosition();

                    while(reader.next(key,value)){
                        String syncSeen = reader.syncSeen()? "*" : "";
                        System.out.printf("[%s%s]\tkey:%s\tvalue:%s\n", position, syncSeen, key, value);
                        rdts.put(key.get(), value.get());
                        position = reader.getPosition();
                    }
                }
            }

            return rdts;
        }catch(IOException ex){
            ex.printStackTrace();
        }
        finally{
            IOUtils.closeStream(reader);
        }

        return null;
    }

}

